package com.xujian.cdc.schema;

import com.xujian.common.beans.DatabaseMapping;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;

/**
 * @author star xu
 * @date 2023/1/30 15:31
 * @Description:
 * @Slogan: 致敬大师, 致敬未来的自己
 */
public class MyKafkaDeserializationSchema implements KafkaDeserializationSchema<DatabaseMapping> {
    @Override
    public boolean isEndOfStream(DatabaseMapping databaseMapping) {
        return false;
    }

    @Override
    public DatabaseMapping deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) {
        String topic = consumerRecord.topic();
        String s = new String(consumerRecord.value());
        return new DatabaseMapping(topic,s);
    }

    @Override
    public TypeInformation<DatabaseMapping> getProducedType() {
        return TypeInformation.of(new TypeHint<DatabaseMapping>() {
            @Override
            public TypeInformation<DatabaseMapping> getTypeInfo() {
                return super.getTypeInfo();
            }
        });
    }
}
